#include "SonyStream.h"

#include "x_socket.h"
#include "x_msg_queue.h"
#include "x_media_msg.h"

#define RECV_SIZE (1024 * 1024)
CSonyStream::CSonyStream(void *pTCPSocket, std::string resid)
: m_recvThread(0)
, m_bStartup(false)
, m_pRecvBuff(NULL)
, m_resid(resid)
{
	m_pTCPSocket = pTCPSocket;
	if (NULL == m_pRecvBuff)
		m_pRecvBuff = new char[RECV_SIZE];

	m_parser.Init();
	J_OS::LOGINFO("CSonyStream::CSonyStream create this = %d", this);
}

CSonyStream::~CSonyStream()
{
	m_parser.Deinit();
	if (m_pRecvBuff != NULL)
	{
		delete m_pRecvBuff;
		m_pRecvBuff = NULL;
	}

	J_OS::LOGINFO("CSonyStream::~CSonyStream destroy this = %d", this);
}

int CSonyStream::Startup()
{
	if (m_bStartup)
		return J_OK;

	m_bStartup = true;
	pthread_create(&m_recvThread, NULL, CSonyStream::WorkThread, this);
	J_OS::LOGINFO("CSonyStream::Startup Startup this = %d", this);

	return J_OK;
}

int CSonyStream::Shutdown()
{
	if (!m_bStartup)
		return J_OK;

	m_bStartup = false;
	pthread_cancel(m_recvThread);
	pthread_join(m_recvThread, NULL);
	J_OS::LOGINFO("CSonyStream::Shutdown Shutdown this = %d", this);

	return J_OK;
}

void CSonyStream::OnWork()
{
	int nLen = 0;
	pthread_setcancelstate(PTHREAD_CANCEL_DEFERRED, NULL);
	while (true)
	{
		nLen = ((J_OS::CTCPSocket *)m_pTCPSocket)->Read(m_pRecvBuff, RECV_SIZE);
		if (nLen < 0)
		{
			CMediaMessage *pMessage = new CMediaMessage();
			pMessage->SetMsgType(MSG_MEDIA_BROKEN);
			pMessage->SetResid(m_resid.c_str());
			CXMessageQueue::Instance()->PostMessage(MSG_MEDIA, pMessage);

			J_OS::LOGERROR("CSonyStream::OnWork recv data error");
			break;
		}

		m_parser.InputData(m_pRecvBuff, nLen);

		int nRet = 0;
		do
		{
			J_StreamHeader streamHeader;
			nRet = m_parser.GetOnePacket(m_pRecvBuff, streamHeader);
			if (nRet == J_OK)
			{
				TLock(m_vecLocker);
				std::vector<CRingBuffer *>::iterator it = m_vecRingBuffer.begin();
				for (; it != m_vecRingBuffer.end(); it++)
				{
					//J_OS::LOGINFO("nDataLen > 0 socket = %d", m_nSocket);
					(*it)->PushBuffer(m_pRecvBuff, streamHeader);
				}
				TUnlock(m_vecLocker);
			}
		}while (nRet == J_OK);
		pthread_testcancel();
	}

	J_OS::LOGINFO("CSonyStream::OnWork pthread_exit");
}
